package com.data.source.api

import org.apache.spark.sql.SparkSession

/**
 * 合并不同schema的qarquet文件
 * 见spark的example: examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala
 */
object SchemaMergeApp {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("schemaMerge").master("local[2]").getOrCreate()

    // This is used to implicitly convert an RDD to a DataFrame.
    import spark.implicits._

    // Create a simple DataFrame, store into a partition directory
    val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
    squaresDF.write.parquet("data/test_table/key=1")

    // Create another DataFrame in a new partition directory,
    // adding a new column and dropping an existing column
    val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
    cubesDF.write.parquet("data/test_table/key=2")

    // Read the partitioned table
    // 需要手动打开mergeSchema的配置，默认是关闭的，因为该功能很耗资源
    val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
    mergedDF.printSchema()

    // The final schema consists of all 3 columns in the Parquet files together
    // with the partitioning column appeared in the partition directory paths
    // root
    //  |-- value: int (nullable = true)
    //  |-- square: int (nullable = true)
    //  |-- cube: int (nullable = true)
    //  |-- key: int (nullable = true)
  }
}
